package main

import (
	"context"
	"fmt"
	"log"

	"github.com/segmentio/kafka-go"
)

func main() {
	broker := "localhost:9092"
	topic := "my-topic"

	for i := 0; i < 10; i++ {

		go func(i int) {

			r := kafka.NewReader(kafka.ReaderConfig{
				Brokers: []string{broker},
				GroupID: "test-consumer-group",
				Topic:   topic,
			})

			fmt.Println("开始:", i)

			for {
				// 从 Kafka 获取消息
				msg, err := r.FetchMessage(context.Background())
				if err != nil {
					log.Println("Error fetching message:", err)
					break
				}

				// 输出消息信息
				log.Printf("Received message: %s from partition: %d", string(msg.Value), msg.Partition)

				fmt.Println(msg)

				// 提交消息
				if err := r.CommitMessages(context.Background(), msg); err != nil {
					log.Println("Error committing message:", err)
				}
			}

		}(i)
	}

	select {}
}
